package com.dxysun.springboot.kafka;

import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.kafka.annotation.KafkaListener;
import org.springframework.kafka.support.Acknowledgment;
import org.springframework.stereotype.Component;

import java.util.Optional;

/**
 * @author dongxiyan
 * @date 2020/3/16 22:36
 */
@Component
public class KafkaReceiver {

    private static Logger logger = LoggerFactory.getLogger(KafkaReceiver.class);

    @Value("${dxy.group.id}")
    private String group;

    @KafkaListener(topics = {"${dxy.topic}"})
    public void listen(ConsumerRecord<?, ?> record, Acknowledgment acknowledgment)
    {
        Optional<?> kafkaMessage = Optional.ofNullable(record.value());
        if (kafkaMessage.isPresent())
        {
            Object message = kafkaMessage.get();
            logger.info("-------------KafkaReceiver receive group="+group+" record=" + record.toString());
        }
        acknowledgment.acknowledge();

    }


}
